package realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.PropertyNamingStrategy;
import com.alibaba.fastjson.serializer.SerializeConfig;
import org.apache.doris.flink.sink.DorisSink;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.*;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import realtime.app.func.DimAsyncFunction;
import realtime.beans.TradeCourseIdOrderBean;
import realtime.beans.TradeOrderBean;
import realtime.util.DateFormatUtil;
import realtime.util.DorisUtil;
import realtime.util.EnvUtil;
import realtime.util.KafkaUtil;

import java.math.BigDecimal;
import java.time.Duration;
import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * @Create 2023/9/8 15:17
 *  下单情况统计
 */
public class DwsTradeSubCateCouOrderWindow {
    public static void main(String[] args) {


        //TODO 1、设置环境
        StreamExecutionEnvironment env = EnvUtil.getSEE(4);

        DataStreamSource<String> sourceDS = env.fromSource(KafkaUtil.getKafkaConsumer("dwd_trade_order_detail", "dws_DTS"),
                WatermarkStrategy.noWatermarks(), "dwd_trade_order_detail"
        );

        //TODO 2、etl
        SingleOutputStreamOperator<JSONObject> JsonObjDS = sourceDS.flatMap(new FlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String value, Collector<JSONObject> out) throws Exception {
                try {
                    JSONObject jsonObject = JSON.parseObject(value);
                    out.collect(jsonObject);
                } catch (Exception e) {
                    e.printStackTrace();
                }

            }
        }).assignTimestampsAndWatermarks(WatermarkStrategy.<JSONObject>forBoundedOutOfOrderness(Duration.ZERO)
                .withTimestampAssigner((value,ts)->value.getLong("ts")*1000));

        //不用对数据去重,没有使用left join

        //TODO 3、按order_id聚合,转换为包装类
        SingleOutputStreamOperator<TradeCourseIdOrderBean> processDS = JsonObjDS.keyBy(value -> value.getString("user_id"))
                .process(new KeyedProcessFunction<String, JSONObject, TradeCourseIdOrderBean>() {

                    ValueState<String> valueState;
                    ValueState<String> userState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        //设置状态的生存时间
                        StateTtlConfig ttlConfig = StateTtlConfig.newBuilder(Time.days(1))
                                .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite) //更新类型
                                .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired) //状态可见性(状态过期之后是否可读)
                                .build();
                        ValueStateDescriptor<String> valueStateDescriptor = new ValueStateDescriptor<String>("state", String.class);
                        valueStateDescriptor.enableTimeToLive(ttlConfig);

                        valueState = getRuntimeContext().getState(valueStateDescriptor);

                        StateTtlConfig ttlConfig1 = StateTtlConfig.newBuilder(Time.seconds(5))
                                .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite) //更新类型
                                .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired) //状态可见性(状态过期之后是否可读)
                                .build();
                        ValueStateDescriptor<String> valueStateDescriptor1 = new ValueStateDescriptor<String>("state1", String.class);
                        valueStateDescriptor.enableTimeToLive(ttlConfig1);

                        userState = getRuntimeContext().getState(valueStateDescriptor1);


                    }

                    @Override
                    public void processElement(JSONObject value, KeyedProcessFunction<String, JSONObject, TradeCourseIdOrderBean>.Context ctx, Collector<TradeCourseIdOrderBean> out) throws Exception {
                        BigDecimal finalAmount = value.getBigDecimal("final_amount");
                        String userId = value.getString("user_id");
                        String courseId = value.getString("course_id");
                        String courseName = value.getString("course_name");
                        Long ts = value.getLong("ts");
                        String toDate = DateFormatUtil.toDate(ts * 1000);

                        TradeCourseIdOrderBean courseIdOrderBean = new TradeCourseIdOrderBean();

                        courseIdOrderBean.setOrderCt(1L);
                        courseIdOrderBean.setUserId(userId);
                        courseIdOrderBean.setOrderAmount(finalAmount);
                        courseIdOrderBean.setCourseId(courseId);
                        courseIdOrderBean.setCourseName(courseName);

                        if (valueState.value() == null) {
                            courseIdOrderBean.setUserCt(1L);
                            valueState.update(toDate);
                        } else if (!valueState.value().equals(DateFormatUtil.toDate(value.getLong("ts")*1000))) {
                            courseIdOrderBean.setUserCt(1L);
                            valueState.update(toDate);
                        } else {
                            courseIdOrderBean.setUserCt(0L);
                        }

                        if (userState.value()==null){
                            courseIdOrderBean.setUserUniqeCt(1L);
                            userState.update(toDate);
                        }else {
                            courseIdOrderBean.setUserUniqeCt(0L);
                        }

                        out.collect(courseIdOrderBean);

                    }

                });


        //TODO 4、按course_id聚合统计
        SingleOutputStreamOperator<TradeCourseIdOrderBean> reduceDS = processDS.keyBy(TradeCourseIdOrderBean::getCourseId)
                .window(TumblingProcessingTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(5)))
                .reduce(new ReduceFunction<TradeCourseIdOrderBean>() {
                    @Override
                    public TradeCourseIdOrderBean reduce(TradeCourseIdOrderBean value1, TradeCourseIdOrderBean value2) throws Exception {
                        value1.setOrderCt(value1.getOrderCt() + value2.getOrderCt());
                        value1.setUserCt(value1.getUserCt() + value2.getUserCt());
                        value1.setUserUniqeCt(value1.getUserUniqeCt()+ value2.getUserUniqeCt());
                        value1.setOrderAmount(value1.getOrderAmount().add(value2.getOrderAmount()));
                        return value1;
                    }
                }, new WindowFunction<TradeCourseIdOrderBean, TradeCourseIdOrderBean, String, TimeWindow>() {
                    @Override
                    public void apply(String s, TimeWindow window, Iterable<TradeCourseIdOrderBean> input, Collector<TradeCourseIdOrderBean> out) throws Exception {
                        TradeCourseIdOrderBean courseIdOrderBean = input.iterator().next();
                        long start = window.getStart();
                        long end = window.getEnd();
                        courseIdOrderBean.setStt(DateFormatUtil.toYmdHms(start));
                        courseIdOrderBean.setEdt(DateFormatUtil.toYmdHms(end));
                        courseIdOrderBean.setCurDate(DateFormatUtil.toDate(start));
                        out.collect(courseIdOrderBean);
                    }
                });

        //reduceDS.print();

        reduceDS = AsyncDataStream.unorderedWait(reduceDS, new DimAsyncFunction<TradeCourseIdOrderBean>("dim_course_info") {
            @Override
            public String getKey(TradeCourseIdOrderBean bean) {
                return bean.getCourseId();
            }

            @Override
            public void join(TradeCourseIdOrderBean bean, JSONObject dim) {
                String subjectId = dim.getString("subject_id");
                bean.setSubjectId(subjectId);
            }
        }, 60, TimeUnit.SECONDS);

        reduceDS = AsyncDataStream.unorderedWait(reduceDS, new DimAsyncFunction<TradeCourseIdOrderBean>("dim_base_subject_info") {
            @Override
            public String getKey(TradeCourseIdOrderBean bean) {
                return bean.getSubjectId();
            }

            @Override
            public void join(TradeCourseIdOrderBean bean, JSONObject dim) {
                String subjectName = dim.getString("subject_name");
                String categoryId = dim.getString("category_id");
                bean.setSubjectName(subjectName);
                bean.setCategoryId(categoryId);
            }
        }, 60, TimeUnit.SECONDS);

        reduceDS = AsyncDataStream.unorderedWait(reduceDS, new DimAsyncFunction<TradeCourseIdOrderBean>("dim_base_category_info") {
            @Override
            public String getKey(TradeCourseIdOrderBean bean) {
                return bean.getCategoryId();
            }

            @Override
            public void join(TradeCourseIdOrderBean bean, JSONObject dim) {
                String categoryName = dim.getString("category_name");
                bean.setCategoryName(categoryName);
            }
        }, 60, TimeUnit.SECONDS);

        SingleOutputStreamOperator<String> endDS = reduceDS.map(value -> {
            SerializeConfig serializeConfig = new SerializeConfig();
            serializeConfig.setPropertyNamingStrategy(PropertyNamingStrategy.SnakeCase);
            String jsonString = JSONObject.toJSONString(value,serializeConfig);
            return jsonString;
        });
        //endDS.print();

        //TODO 14、数据写入doris中
        DorisSink<String> dorisSink = DorisUtil.getDorisSink("dws_trade_sub_cate_cou_order_window");

        endDS.sinkTo(dorisSink);

        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}
